package com.clx.core;

import com.clx.CrpcBootstrap;
import com.clx.NettyBootstrapInitializer;
import com.clx.compress.CompressorFactory;
import com.clx.discovery.Registry;
import com.clx.enumeration.RequestType;
import com.clx.serialize.SerializerFactory;
import com.clx.transport.message.CrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 心跳检测：探测服务器连接可用性，异常时重试（最多3次）
 */
@Slf4j
public class HeartbeatDetector {

    // 记录每个服务地址的重试次数
    private static final Map<InetSocketAddress, Integer> RETRY_TIMES_CACHE = new ConcurrentHashMap<>();

    public static void detectorHeartbeat(String serviceName) {
        // 从注册中心拉取服务列表
        Registry registry = CrpcBootstrap.getInstance().getRegistry();
        List<InetSocketAddress> addresses = registry.lookup(serviceName);

        // 初始化连接与重试次数
        for (InetSocketAddress address : addresses) {
            try {
                if (!CrpcBootstrap.CHANNEL_CACHE.containsKey(address)) {
                    Channel channel = NettyBootstrapInitializer.getBootstrap()
                            .connect(address)
                            .sync()
                            .channel();
                    CrpcBootstrap.CHANNEL_CACHE.put(address, channel);
                    RETRY_TIMES_CACHE.put(address, 0); // 新连接重试次数初始化为0
                }
            } catch (InterruptedException e) {
                throw new RuntimeException("初始化服务连接失败", e);
            }
        }

        // 启动定时心跳任务（守护线程，随主程序退出）
        Thread heartbeatThread = new Thread(() ->
                new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 2000),
                "clx-heartbeat-detector"
        );
        heartbeatThread.setDaemon(true);
        heartbeatThread.start();
    }

    private static class MyTimerTask extends TimerTask {

        @Override
        public void run() {
            // 清空本次响应时间统计
            CrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            // 遍历所有缓存的Channel（ConcurrentHashMap保证线程安全）
            for (Map.Entry<InetSocketAddress, Channel> entry : CrpcBootstrap.CHANNEL_CACHE.entrySet()) {
                InetSocketAddress address = entry.getKey();
                Channel channel = entry.getValue();

                // 跳过已关闭的通道
                if (!channel.isActive()) {
                    log.warn("通道[{}]已关闭，移除缓存", address);
                    CrpcBootstrap.CHANNEL_CACHE.remove(address);
                    RETRY_TIMES_CACHE.remove(address);
                    continue;
                }

                long startTime = System.currentTimeMillis();
                // 构建心跳请求
                CrpcRequest heartBeatReq = CrpcRequest.builder()
                        .requestId(CrpcBootstrap.ID_GENERATOR.getId())
                        .compressType(CompressorFactory.getCompressor(CrpcBootstrap.COMPRSS_TYPE).getCode())
                        .requestType(RequestType.HEART_BEAT.getId())
                        .serializeType(SerializerFactory.getSerialze(CrpcBootstrap.SERIALIZE_TYPE).getCode())
                        .timeStamp(startTime)
                        .build();

                // 关联异步结果，用于等待心跳响应
                CompletableFuture<Object> future = new CompletableFuture<>();
                CrpcBootstrap.PENDING_REQUEST.put(heartBeatReq.getRequestId(), future);

                // 发送心跳并监听“写操作结果”
                channel.writeAndFlush(heartBeatReq).addListener((ChannelFutureListener) writeFuture -> {
                    if (!writeFuture.isSuccess()) {
                        log.error("向[{}]写心跳请求失败", address, writeFuture.cause());
                        handleRetry(address, writeFuture.cause()); // 写失败时直接重试
                        future.completeExceptionally(writeFuture.cause());
                    }
                });

                // 监听“响应结果/异常”
                future.whenComplete((result, ex) -> {
                    long responseTime = System.currentTimeMillis() - startTime;
                    if (ex != null) {
                        handleRetry(address, ex); // 响应异常时重试
                    } else {
                        // 心跳成功：重置重试次数 + 记录响应时间
                        RETRY_TIMES_CACHE.put(address, 0);
                        CrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(responseTime, channel);
                        log.debug("与[{}]的心跳响应时间：{}ms", address, responseTime);
                    }
                });
            }

            // 打印响应时间统计
            log.info("----------------------心跳响应时间统计----------------------");
            CrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.forEach(
                    (time, channel) -> log.debug("响应时间[{}ms] -> 通道ID[{}]", time, channel.id())
            );
        }

        /**
         * 统一处理“连接异常重试”逻辑
         */
        private void handleRetry(InetSocketAddress address, Throwable cause) {
            int retryTimes = RETRY_TIMES_CACHE.getOrDefault(address, 0) + 1;
            if (retryTimes <= 3) {
                RETRY_TIMES_CACHE.put(address, retryTimes);
                log.warn("和地址为[{}]的主机连接发生异常，正在进行第[{}]次重试......", address, retryTimes);
            } else {
                log.error("与[{}]的心跳连续失败3次，移除连接！异常原因：{}", address, cause.getMessage());
                CrpcBootstrap.CHANNEL_CACHE.remove(address);
                RETRY_TIMES_CACHE.remove(address);
            }
        }
    }
}